package com.ruby.bigtable.hbase;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MapperReducer extends Configured implements Tool {
	
	public static class MyMapper extends TableMapper<Text, Put>{
		public void map(ImmutableBytesWritable key, Result value,
				Context context) throws IOException, InterruptedException {
			Put put = new Put(key.get());
			for(Cell cell: value.rawCells()){
				if("info".equals(Bytes.toString(CellUtil.cloneFamily(cell)))){
					if("username".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){
						put.add(Bytes.toBytes("info"), Bytes.toBytes("mame"), CellUtil.cloneValue(cell));
					}
				}
			}
			context.write(new Text(Bytes.toString(key.get())), put);
		}
	}
	
	public static class MyReducer extends TableReducer<Text, Put, ImmutableBytesWritable>{
		public void reduce(Text arg0, Iterable<Put> arg1, 
				Context arg2) throws IOException, InterruptedException {
			for(Put put: arg1){
				arg2.write(null, put);
			}
		}
	}

	public int run(String[] args) throws Exception {
		Job job = Job.getInstance(this.getConf(), this.getClass().getSimpleName());
		job.setJarByClass(this.getClass());
		Scan scan = new Scan();
		scan.setCacheBlocks(false);
		scan.setCaching(500);
		TableMapReduceUtil.initTableMapperJob("test:user", scan, MyMapper.class, Text.class, Put.class, job);
		TableMapReduceUtil.initTableReducerJob("test:sub_user", MyReducer.class, job);
		job.setNumReduceTasks(1);
		return job.waitForCompletion(true) ? 0: 1;
	}
	
	public static void main(String[] args) throws Exception {
		Configuration conf = HBaseConfiguration.create();
		System.exit(ToolRunner.run(conf, new MapperReducer(), args));
	}

}
